package cn.baiyang.rocketmq.spring.core;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.apis.producer.Transaction;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
import org.apache.rocketmq.shaded.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * RocketMQ Message Sender
 *
 * @author hongzhu
 * @version V1.0
 * @since 2023-07-05 18:25
 */
public abstract class AbstractRocketMQSender {
    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractRocketMQSender.class);

    protected String topic;

    protected String tag;

    protected String messageGroup;

    protected Producer producer;

    protected AbstractRocketMQSender(String topic) {
        this.topic = topic;
    }

    protected AbstractRocketMQSender(String topic, String tag) {
        this.topic = topic;
        this.tag = tag;
    }

    protected AbstractRocketMQSender(String topic, String tag, String messageGroup) {
        this.topic = topic;
        this.tag = tag;
        this.messageGroup = messageGroup;
    }

    protected abstract void setProducer(Producer producer);

    public Producer getProducer() {
        return producer;
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @return Message Id String
     */
    public String send(String body) {
        return send(body.getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param keys key list
     * @return Message Id String
     */
    public String send(String body, String... keys) {
        return send(body.getBytes(StandardCharsets.UTF_8), null, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @return Message Id String
     */
    public String send(String body, Map<String, String> properties) {
        return send(body.getBytes(StandardCharsets.UTF_8), null, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @return Message Id String
     */
    public String send(String body, Long deliveryTimestamp) {
        return send(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, (Map<String, String>) null, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String send(String body, Map<String, String> properties, String... keys) {
        return send(body.getBytes(StandardCharsets.UTF_8), null, properties, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param keys key list
     * @return Message Id String
     */
    public String send(String body, Long deliveryTimestamp, String... keys) {
        return send(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @return Message Id String
     */
    public String send(String body, Long deliveryTimestamp, Map<String, String> properties) {
        return send(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String send(String body, Long deliveryTimestamp, Map<String, String> properties, String... keys) {
        return send(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, properties, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @return Message Id String
     */
    public String send(byte[] body) {
        return send(body, (Long) null, (Map<String, String>) null, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param keys key list
     * @return Message Id String
     */
    public String send(byte[] body, String... keys) {
        return send(body, null, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @return Message Id String
     */
    public String send(byte[] body, Map<String, String> properties) {
        return send(body, null, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @return Message Id String
     */
    public String send(byte[] body, Long deliveryTimestamp) {
        return send(body, deliveryTimestamp, (Map<String, String>) null, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String send(byte[] body, Map<String, String> properties, String... keys) {
        return send(body, null, properties, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param keys key list
     * @return Message Id String
     */
    public String send(byte[] body, Long deliveryTimestamp, String... keys) {
        return send(body, deliveryTimestamp, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @return Message Id String
     */
    public String send(byte[] body, Long deliveryTimestamp, Map<String, String> properties) {
        return send(body, deliveryTimestamp, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String send(byte[] body, Long deliveryTimestamp, Map<String, String> properties, String... keys) {
        return send(buildMessage(body, deliveryTimestamp, properties, keys));
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @return Message Id String
     */
    public String sendAsync(String body) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(String body, String... keys) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8), null, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @return Message Id String
     */
    public String sendAsync(String body, Map<String, String> properties) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8), null, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @return Message Id String
     */
    public String sendAsync(String body, Long deliveryTimestamp) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, (Map<String, String>) null, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(String body, Map<String, String> properties, String... keys) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8), null, properties, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(String body, Long deliveryTimestamp, String... keys) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @return Message Id String
     */
    public String sendAsync(String body, Long deliveryTimestamp, Map<String, String> properties) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(String body, Long deliveryTimestamp, Map<String, String> properties, String... keys) {
        return sendAsync(body.getBytes(StandardCharsets.UTF_8), deliveryTimestamp, properties, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @return Message Id String
     */
    public String sendAsync(byte[] body) {
        return sendAsync(body, (Long) null, (Map<String, String>) null, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(byte[] body, String... keys) {
        return sendAsync(body, null, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @return Message Id String
     */
    public String sendAsync(byte[] body, Map<String, String> properties) {
        return sendAsync(body, null, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @return Message Id String
     */
    public String sendAsync(byte[] body, Long deliveryTimestamp) {
        return sendAsync(body, deliveryTimestamp, (Map<String, String>) null, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(byte[] body, Map<String, String> properties, String... keys) {
        return sendAsync(body, null, properties, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(byte[] body, Long deliveryTimestamp, String... keys) {
        return sendAsync(body, deliveryTimestamp, null, keys);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @return Message Id String
     */
    public String sendAsync(byte[] body, Long deliveryTimestamp, Map<String, String> properties) {
        return sendAsync(body, deliveryTimestamp, properties, (String) null);
    }

    /**
     * 发送成功后返回message id
     * @param body payload
     * @param deliveryTimestamp delay time
     * @param properties attributes list
     * @param keys key list
     * @return Message Id String
     */
    public String sendAsync(byte[] body, Long deliveryTimestamp, Map<String, String> properties, String... keys) {
        return sendAsync(buildMessage(body, deliveryTimestamp, properties, keys));
    }

    /**
     * 发送成功后返回message id
     * @param message {@link Message}
     * @param transaction {@link Transaction}
     * @throws java.lang.Exception sending exception
     * @return Message Id String
     */
    public String send(Message message, Transaction transaction) throws Exception {
        try {
            SendReceipt sendReceipt = producer.send(message, transaction);
            LOGGER.debug("Send transaction message successfully, messageId={}", sendReceipt.getMessageId());
            return sendReceipt.getMessageId().toString();
        } catch (Exception e) {
            LOGGER.error("Failed to send transaction message", e);
            throw e;
        }
    }

    private Message buildMessage(byte[] body, Long deliveryTimestamp, Map<String, String> properties, String[] keys) {
        MessageBuilder messageBuilder = new MessageBuilderImpl();
        messageBuilder = messageBuilder.setTopic(topic);
        if (StringUtils.isNotBlank(tag)) { // 设置消息Tag，用于消费端根据指定Tag过滤消息
            messageBuilder = messageBuilder.setTag(tag);
        }
        if (null != deliveryTimestamp && StringUtils.isNotBlank(messageGroup)) {
            throw new RuntimeException("请检查消息类型，不能同时发送顺序消息和延迟消息");
        }
        if (StringUtils.isNotBlank(messageGroup)) { // 设置message group来保证顺序消息
            messageBuilder = messageBuilder.setMessageGroup(messageGroup);
        }
        if (null != deliveryTimestamp) { // 设置延迟投递消息的时间，用于延时/定时消息
            messageBuilder = messageBuilder.setDeliveryTimestamp(deliveryTimestamp);
        }
        if (null != keys && keys.length > 0) { // 设置消息索引键，可根据关键字精确查找某条消息
            if (!Arrays.asList(keys).contains(null)){
                messageBuilder = messageBuilder.setKeys(keys);
            }
        }
        if (null != properties && !properties.isEmpty()) { // 设置附加属性，可用于SQL92类型的消息过滤
            for (Map.Entry<String, String> e : properties.entrySet()) {
                messageBuilder = messageBuilder.addProperty(e.getKey(), e.getValue());
            }
        }
        return messageBuilder.setBody(body).build();
    }

    /**
     * 发送成功后返回message id, 或者失败返回空字符
     * @param message {@link Message}
     * @return Message Id String
     */
    private String send(Message message) {
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);
            LOGGER.debug("Send message successfully, messageId={}", sendReceipt.getMessageId());
            return sendReceipt.getMessageId().toString();
        } catch (Exception e) {
            LOGGER.error("Failed to send message", e);
            return "";
        }
    }

    /**
     * 发送成功后返回message id, 或者失败返回空字符<br/>
     * 异步发送时立即返回为空字符
     * @param message {@link Message}
     * @return Message Id String
     */
    private String sendAsync(Message message) {
        final String[] messageId = new String[]{""};
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            CompletableFuture<SendReceipt> completableFuture = producer.sendAsync(message);
            ExecutorService sendCallbackExecutor = Executors.newCachedThreadPool();
            completableFuture.whenCompleteAsync((sendReceipt, throwable) -> {
                if (null != throwable) {
                    LOGGER.error("Failed to send async message on complete", throwable);
                    return;
                }
                messageId[0] = sendReceipt.getMessageId().toString();
                LOGGER.debug("Send async message successfully, messageId={}", messageId[0]);
            }, sendCallbackExecutor);
        } catch (Exception e) {
            LOGGER.error("Failed to send sync message", e);
        }
        return messageId[0];
    }

}
